package com.qyer.log.job.sum;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import com.qyer.util.log.entity.TrackLog;
import com.qyer.util.log.service.LogService;
import com.qyer.util.log.service.impl.LogServiceImpl;
import com.qyer.util.log.util.CateUtil;
import com.qyer.util.log.util.LogFatory;
import com.qyer.util.log.util.MyDateUtil;
import com.qyer.util.log.util.StatisticsUtils;
import com.qyer.util.log.util.TrackSourceType;



public class UserTracksSummary {
    
	public static class M1 extends Mapper<LongWritable, Text, Text, Text> {
		 
		static HashMap<String, String> sourceMap= new HashMap<String, String>();
		static Map<String, String> cateMap= new LinkedHashMap<String, String>();
		
		protected void setup(Context context) throws IOException, InterruptedException {
			BufferedReader in = null;

			try {
				// 从当前作业中获取要缓存的文件
				Path[] paths = DistributedCache.getLocalCacheFiles(context.getConfiguration());
				
				System.out.println("paths===="+paths);
				String line = null;
				String[] strArr = null;
				for (Path path : paths) {
					System.out.println("path===="+path.toString());
					
					if (path.toString().contains("sourcetype.txt")) {
						in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()),"utf-8")));
						while (null != (line = in.readLine())) {
							strArr = line.split("\t", -1);
							
							if(strArr.length<4){
								continue ;
							}
							
							if("3".equals(strArr[0])){
								sourceMap.put(strArr[4], strArr[0]+"@"+strArr[1]);
							}
						}
						
					}else if (path.toString().contains("cate_business.txt")) {
						in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()),"utf-8")));
						while (null != (line = in.readLine())) {
							strArr = line.split("\t", -1);
							
							if(strArr.length<4){
								continue ;
							}
							
							cateMap.put(strArr[1]+","+strArr[2], strArr[3]);
						}
					}
				}
				
				if (in != null) {
					in.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			} finally {
				try {
					if (in != null) {
						in.close();
					}
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		
		private Text val = new Text();
		private Text keyVal =new Text();
		
		final static String FS="\t";
		
		//清洗日志工具类
		LogService lli = (LogService)LogFatory.getInstance(LogServiceImpl.class.getName());
		
		TrackLog tracklog = null ;
		
		public void map(LongWritable key, Text value, Context context) throws IOException,InterruptedException {
			
			tracklog= lli.getTrackFormat(value.toString());
			
			if(tracklog==null){
				return ; 
			}
			
			String becon_type=tracklog.getType();//获取becontype
			
			if(!"1".equals(becon_type)){
				return ; 
			}
			String url=tracklog.getCurUrl();//获取url地址
			String platform = tracklog.getPlatform(); 
			
			if("mobile".equals(platform)){
				platform="m";
			}
			
			if("web".equals(platform)||"m".equals(platform)){
			
				String unique_id=tracklog.getUnique_id();//获取唯一标示
				String refer=tracklog.getRefUrl();//获取refer
				String os_browser =tracklog.getUserAgent();//获取平台
				
				String sourcetype =TrackSourceType.getSourceType(refer, url, sourceMap, cateMap, os_browser); 
				String cate =CateUtil.getCate(url, cateMap);
				
			    
			    for (String p : new String []{"A" , platform }){
			    	for (String c : new String []{"A" , cate }){
			    		for(String sou : StatisticsUtils.getDimListTSource(sourcetype)) {
			    			
			    			sou=sou.replace("@", FS);
			    			
			    			String keyStr = p+FS+c+FS+sou; 
			    			String valStr = unique_id;
			    			keyVal.set(keyStr);
			    			val.set(valStr);
			    			context.write(keyVal,val);    
						}
				    }
			    }
				
			}		
		}         
	 }
	 
	 
		 public static class Combiner extends Reducer<Text, Text, Text, Text> {

		  private Text val = new Text();
		  private Text keyVal =new Text();
		  
		  final static String FS="\t";
		  
		  String DATE = "";
	    
	     @Override
	     public void reduce(Text key, Iterable<Text> values, Context context) throws IOException,InterruptedException {
	    	 	String[] keyStrs = key.toString().split(FS); 
	    	 	if(keyStrs.length<4){
	    	 		return ; 
	    	 	}
	    	 	
	    	 	//去掉站内来源 本业务线内的来源维度
	    	 	if(keyStrs[2].equals("2")&&!"A".equals(keyStrs[1])){
	    	 		if(keyStrs[1].equals(keyStrs[3])){
	    	 			return ; 
	    	 		}
	    	 		
	    	 		//去掉全业务线  站内来源数据
//	    	 		if("A".equals(keyStrs[1])){
//	    	 			return ;
//	    	 		}
	    	 	}
	    	 	
				Set<String> cookieIdSet = new HashSet<String>();
				long pv = 0l;
	
				for (Text valT : values) {
					pv++; 
					cookieIdSet.add(valT.toString()); 
				}
	
				keyVal.set(key);
				val.set(pv + FS + cookieIdSet.toString());
//				cookieIdSet.clear();
				context.write(keyVal, val);       
	     }
	 }

	 
	 public static class R1 extends Reducer<Text, Text, Text, Text> {
    
    	  private Text val = new Text();
 		  private Text keyVal =new Text();
 		  
 		  final static String FS="\t";
 		  
 		  static String DATE ;
 		  
 		  static String len;
 		  
         
 		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException,InterruptedException {
  			
 			if (DATE == null) {
				DATE = context.getConfiguration().get("DATE");
			} 
 			
 			if (len == null) {
 				len = context.getConfiguration().get("len");
			}
 			
  			Set<String> cookieIdSet = new HashSet<String>();
  			long uv = 0l;
  			long pv = 0l;

  			for (Text valT : values) {
				String val  = valT.toString(); 
				String pvStr =val.split(FS)[0];
				
				pv=pv+Long.parseLong(pvStr);
				
				String uvStr =val.split(FS)[1];
				
				uvStr=uvStr.replace("[", "");
				uvStr=uvStr.replace("]", "");
				
				if(uvStr.indexOf(",")>-1){
					for(String uvS : uvStr.split(",")){
						cookieIdSet.add(uvS); 
					}
				}else{
					cookieIdSet.add(uvStr); 
				}
  			}

  			uv=cookieIdSet.size();
  			
  			
  			SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
  			Long curTimes=0L;
			try {
				String d =sdf.format(new Date());
				curTimes = sdf.parse(d).getTime();
			} catch (ParseException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			curTimes=curTimes/1000;
			
  			keyVal.set(curTimes+FS +DATE+FS+uv+FS+len);
  			val.set(key+FS+cookieIdSet.toString());
  			context.write(keyVal, val);       
          }
         
     }
     /**
     * @param args
     * @throws URISyntaxException 
     * @throws ParseException 
     * @throws InterruptedException 
     * @throws ClassNotFoundException 
     */
     public static void main(String[] args) throws IOException, URISyntaxException, ParseException, ClassNotFoundException, InterruptedException{
        
		Configuration conf = new Configuration();
//		conf.set("mapreduce.reduce.memory.mb" ,"3048");
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		int exitCode = 127;	 
    	 
	    for(String arg : args ){
	    	System.out.println(" arg  is " +arg);
	    }
	    
		if (otherArgs.length < 2) {
			System.out.println("************************************************************");
			System.out.println("************************************************************");
			System.out.println("Usage: please input 1 params, for example: file.jar args[0] args[1]");
			System.out.println("args[0] is dateList: 20130101,20130106   or 20130106");
			System.out.println("args[1] is day or week or  mouth ");
			System.out.println("************************************************************");
			System.out.println("************************************************************");
			System.exit(exitCode);
		}
          
		String startDate = otherArgs[0].split(",")[0];
		String endDate = otherArgs[0].split(",").length == 2 ? otherArgs[0].split(",")[1] : startDate;
		
		String timeCycle = otherArgs[1]; 
		
		String baseWeb = "/flume/clean/track/";
		String baseOutPath = "/flume/summary/UserTracksSummary/"+timeCycle;
		
		String cateMapPath ="/flume/dict/cate_business/20151020";
		String sourcetypePath="/flume/dict/sourcetype/20151020";
		
		
		//单天
		if ("day".equalsIgnoreCase(timeCycle)){
			SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
			Long startDateLong =sdf.parse(startDate).getTime();
			startDateLong=startDateLong/1000;
			conf.set("DATE", startDateLong.toString());
			conf.set("len", "1");
		
		
			Job job = new Job(conf, "UserTracksSummary");
			job.setInputFormatClass(TextInputFormat.class);
			job.setJarByClass(UserTracksSummary.class);
			job.setMapperClass(M1.class);
			job.setCombinerClass(Combiner.class);
			job.setReducerClass(R1.class);
			job.setNumReduceTasks(3);
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(Text.class);
			job.setMapOutputKeyClass(Text.class);
			job.setMapOutputValueClass(Text.class);
		
	        
	        // source type 字典
 			FileSystem hdfs1 = FileSystem.get(conf);
 			FileStatus[] fileList = hdfs1.listStatus(new Path(sourcetypePath));
 			for (int i = 0; i < fileList.length; i++) {
 				String fileName = fileList[i].getPath().getName();
				if (fileName.contains("sourcetype.txt")) {
					System.out.println("----------------->>> " + sourcetypePath + "/" + fileName);
					String inPath2Link = new Path(sourcetypePath + "/" + fileName).toUri().toString() + "#"+ "sourcetype.txt" + i;
					DistributedCache.addCacheFile(new URI(inPath2Link), job.getConfiguration());
				}
 			}
 			
 			
 			// category busniess字典
 			hdfs1 = FileSystem.get(conf);
 			fileList = hdfs1.listStatus(new Path(cateMapPath));
 			for (int i = 0; i < fileList.length; i++) {
 				String fileName = fileList[i].getPath().getName();
				if (fileName.contains("cate_business.txt")) {
					System.out.println("----------------->>> " + cateMapPath + "/" + fileName);
					String inPath2Link = new Path(cateMapPath + "/" + fileName).toUri().toString() + "#"+ "cate_business.txt" + i;
					DistributedCache.addCacheFile(new URI(inPath2Link), job.getConfiguration());
				}
 			}
	        
	        
 			for (String rDate : MyDateUtil.getDateList(startDate, endDate, 0)) {
 				System.out.println("runDate -------------->" +rDate);
 				FileInputFormat.addInputPath(job, new Path(baseWeb + rDate));
 			}
	        
	        String outPath = baseOutPath +"/"+ startDate;
	        FileSystem.get(conf).delete(new Path(outPath), true);
	        FileOutputFormat.setOutputPath(job, new Path(outPath));
	        
	        exitCode = job.waitForCompletion(true) ? 0 : 1;
	        
	        System.out.println("--------------------------------------------END1"+startDate);
		        
		}
		
     }
}
